/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server;

import java.io.Flushable;
import java.io.IOException;
import java.util.LinkedList;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 在ProposalRequestProcessor处理器中，后续会有两个处理器CommitProcessor和SyncRequestProcessor.
 * 如果是事务请求，会经过SyncRequestProcessor.
 *
 * SyncRequestProcessor:事务日志记录处理器。将事务请求写入到日志文件中进行持久化，并在写入的日志过多会进行对当前的数据进行快照操作的处理,
 * 可以看到它快照的时机是[snapCount/2,snapCount]中的一个随机数，主要是防止所有的服务器同时进行快照操作。
 *
 * 这个处理器也是用一个单线程来处理加入进来的请求，
 * 对于事务请求会写入到磁盘缓冲，但是没有进行flush操作，
 * 其flush操作主要是等到后续没有事务请求加入进来或者写入的数据量过大则会进行一次flush操作，
 * 其flush操作主要是将日志从磁盘缓冲刷入到磁盘进行持久化，并且将等待刷新的事务请求交给后续的处理器进行处理。
 */
public class SyncRequestProcessor extends Thread implements RequestProcessor {
    private static final Logger LOG = LoggerFactory.getLogger(SyncRequestProcessor.class);
    private final ZooKeeperServer zks;
    /**
     * 请求队列
     */
    private final LinkedBlockingQueue<Request> queuedRequests = new LinkedBlockingQueue<>();
    /**
     * 下一个处理器
     */
    private final RequestProcessor nextProcessor;
    // 处理快照的线程
    private Thread snapInProcess = null;
    // 是否在运行
    volatile private boolean running;

    /**
     * 等待被刷到磁盘的请求队列
     */
    private final LinkedList<Request> toFlush = new LinkedList<>();
    private final Random r = new Random(System.nanoTime());
    // 快照的个数
    private static int snapCount = ZooKeeperServer.getSnapCount();

    private final Request requestOfDeath = Request.requestOfDeath;

    public SyncRequestProcessor(ZooKeeperServer zks, RequestProcessor nextProcessor) {
        super("SyncThread:" + zks.getServerId());
        this.zks = zks;
        this.nextProcessor = nextProcessor;
        running = true;
    }

    /**
     * used by tests to check for changing
     * snapcounts
     * @param count
     */
    public static void setSnapCount(int count) {
        snapCount = count;
    }

    /**
     * used by tests to get the snapcount
     * @return the snapcount
     */
    public static int getSnapCount() {
        return snapCount;
    }

    /**
     * 核心方法，消费请求队列,批处理进行快照以及刷到事务日志
     *
     * run方法注意的点
     * 1.randRoll的意义，可以看到语句 if (logCount > (snapCount / 2 + randRoll))
     *     这是用来判断logCount是否足够，如果足够了，代表一个事务日志记录的量够了，
     *     下面调用rollLog，就会生成下一个事务日志文件了。
     * 2.“批”处理的思想
     *     当 logCount > (snapCount / 2 + randRoll) 时，批处理的思想
     *     对于事务日志，此时才调用rollLog写入到下一个事务日志
     *     对于快照，如果可行，就调用zks.takeSnapshot()进行快照
     *
     * 而不是每一个请求就一个事务日志，不是每一个请求就生成一次快照
     * 然后对于事务日志，当toFlush.size() > 1000才会调用flush函数
     *
     * SyncRequestProcessor会尽快的用while循环，把积压在内存队列里排队的proposal的请求全部都尽快的写入到磁盘上的事务日志里去。
     */
    @Override
    public void run() {
        try {
            int logCount = 0;
            // randRoll是一个 snapCount/2以内的随机数, 避免所有机器同时进行snapshot
            int randRoll = r.nextInt(snapCount/2);
            while (true) {
                Request si = null;
                //没有要刷到磁盘的请求
                if (toFlush.isEmpty()) {
                    si = queuedRequests.take(); //没有数据会在此阻塞
                } else {
                    //有需要刷到磁盘的请求
                    si = queuedRequests.poll(); // 没有数据不会进行阻塞
                    //触发flush()的有2个条件,条件1：如果请求队列的当前请求为空
                    if (si == null) {
                        //刷到磁盘
                        flush(toFlush);
                        continue;
                    }
                }

                //如果是结束标识请求, 则停止当前线程
                if (si == requestOfDeath) {
                    break;
                }
                //请求队列取出了请求
                if (si != null) {
                    //核心方法：写入内存数据库， 请求添加至日志文件，只有事务性请求才会返回true
                    if (zks.getZKDatabase().append(si)) {
                        logCount++;
                        //如果写入的日志数量到了一定的量, 则对当前数据进行快照操作
                        if (logCount > (snapCount / 2 + randRoll)) {
                            randRoll = r.nextInt(snapCount/2);
                            //事务日志滚动到另外一个文件记录
                            zks.getZKDatabase().rollLog();// 这里是换一个新的文件来存储事务日志
                            //正在进行快照
                            if (snapInProcess != null && snapInProcess.isAlive()) {
                                LOG.warn("Too busy to snap, skipping");
                            } else {
                                snapInProcess = new Thread("Snapshot Thread") {
                                        @Override
                                        public void run() {
                                            try {
                                                //使用一个新的线程来进行快照操作，对内存的dataTree,session等数据以文件形式存储下来
                                                //进行快照,将sessions和datatree保存至snapshot文件
                                                zks.takeSnapshot();
                                            } catch(Exception e) {
                                                LOG.warn("Unexpected exception", e);
                                            }
                                        }
                                    };
                                //启动线程
                                snapInProcess.start();
                            }
                            //重置
                            logCount = 0;
                        }
                    }
                    //刷到磁盘的队列为空
                    else if (toFlush.isEmpty()) {
                        //下个处理器处理
                        nextProcessor.processRequest(si);
                        if (nextProcessor instanceof Flushable) {
                            //下个处理器可以刷，就刷
                            ((Flushable)nextProcessor).flush();
                        }
                        continue;
                    }
                    //刷的队列添加记录
                    toFlush.add(si);
                    //触发flush()的有2个条件,条件2：超过了1000条就一起刷到磁盘
                    if (toFlush.size() > 1000) {
                        flush(toFlush);
                    }
                }
            }
        } catch (Throwable t) {
            LOG.error("Severe unrecoverable error, exiting", t);
            running = false;
            System.exit(11);
        }
        LOG.info("SyncRequestProcessor exited!");
    }

    /**
     * 批处理的思想，把事务日志刷到磁盘，让下一个处理器处理
     * @param toFlush
     * @throws IOException
     * @throws RequestProcessorException
     */
    private void flush(LinkedList<Request> toFlush) throws IOException, RequestProcessorException {
        //队列为空，没有需要刷的
        if (toFlush.isEmpty()) {
            return;
        }
        //事务日志刷到磁盘
        zks.getZKDatabase().commit();
        // 将等待后续处理器处理的请求调用后续处理器进行处理
        while (!toFlush.isEmpty()) {
            Request i = toFlush.remove();
            //下一个处理器处理
            nextProcessor.processRequest(i);
        }
        // 这里主要是对SendAckRequestProcessor这些需要进行IO操作的处理器进行多个事务一起flush以提升效率的操作
        if (nextProcessor instanceof Flushable) {
            //下个处理器也可以刷，就刷
            ((Flushable)nextProcessor).flush();
        }
    }

    /**
     * 队列添加requestOfDeath请求，线程结束后，调用flush函数，最后关闭nextProcessor
     */
    @Override
    public void shutdown() {
        LOG.info("Shutting down");
        queuedRequests.add(requestOfDeath);
        try {
            if(running){
                this.join();
            }
        } catch(InterruptedException e) {
            LOG.warn("Interrupted while wating for " + this + " to finish");
        }
        nextProcessor.shutdown();
    }

    /**
     * 生产者，加入请求队列
     * @param request
     */
    @Override
    public void processRequest(Request request) {
        queuedRequests.add(request);
    }

}
